// Copyright 2015 The etcd Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package raft

import (
	"fmt"
	"github.com/pingcap-incubator/tinykv/log"
	pb "github.com/pingcap-incubator/tinykv/proto/pkg/eraftpb"
	"sync"
)

// RaftLog manage the log entries, its struct look like:
//
//  snapshot/first.....applied....committed....stabled.....last
//  --------|------------------------------------------------|
//                            log entries
//
// for simplify the RaftLog implement should manage all log entries
// that not truncated
type RaftLog struct {
	id uint64
	// storage contains all stable entries since the last snapshot.
	storage Storage

	// committed is the highest log position that is known to be in
	// stable storage on a quorum of nodes.
	committed uint64

	// applied is the highest log position that the application has
	// been instructed to apply to its state machine.
	// Invariant: applied <= committed
	applied uint64

	// log entries with index <= stabled are persisted to storage.
	// It is used to record the logs that are not persisted by storage yet.
	// Everytime handling `Ready`, the unstabled logs will be included.
	stabled uint64

	// all entries that have not yet compact.
	entries []pb.Entry

	// the incoming unstable snapshot, if any.
	// (Used in 2C)
	pendingSnapshot *pb.Snapshot

	// Your Data Here (2A).

	rwLock sync.RWMutex
}

// newLog returns log using the given storage. It recovers the log
// to the state that it just commits and applies the latest snapshot.
func newLog(storage Storage) *RaftLog {
	// Your Code Here (2A).
	initial, _, _ := storage.InitialState()

	raftLog := &RaftLog{}
	raftLog.storage = storage
	raftLog.stabled = 0
	raftLog.committed = initial.Commit
	raftLog.applied = 0
	raftLog.entries = make([]pb.Entry, 0) // 第一个Log是NoopLog
	log.Infof("[RaftLog] init from storage. committed = %d, applied = %d", raftLog.committed, raftLog.applied)

	// 处理storage预先有值的场景
	startIndex, fail := raftLog.storage.FirstIndex()
	if fail != nil {
		panic("[RaftLog] fail fetch storage start_index")
	}

	stopIndex, fail := raftLog.storage.LastIndex()
	if fail != nil {
		panic("[RaftLog] fail fetch storage stop_index")
	}

	for i := uint64(0); i < startIndex; i++ {
		// 预置[0, start_idx]之间的空Entry
		raftLog.entries = append(raftLog.entries, pb.Entry{Term: 0, Index: i})
	}

	if startIndex > stopIndex {
		raftLog.stabled = startIndex - 1
		log.Infof("[RaftLog] init zero, start_idx: %v, stop_idx: %v. last_index: %v", startIndex, stopIndex, raftLog.LastIndex())
		return raftLog
	}
	log.Info("[RaftLog] init from storage. start_idx: %v, stop_idx: %v", startIndex, stopIndex)

	storageEntries, fail := storage.Entries(startIndex, stopIndex+1) //storage.Entries方法是左闭右开区间，所有stopIndex+1传入
	if fail != nil {
		panic(fmt.Sprintf("[RaftLog] fail init from storage: %v", fail))
	}

	for _, entry := range storageEntries {
		raftLog.entries = append(raftLog.entries, entry)
	}
	raftLog.stabled = stopIndex
	log.Info("[RaftLog] init in storage. stable: %+v", raftLog.stabled)
	return raftLog
}

func (l *RaftLog) setRaftIndex(idx uint64) {
	l.id = idx
}

// We need to compact the log entries in some point of time like
// storage compact stabled log entries prevent the log entries
// grow unlimitedly in memory
func (l *RaftLog) maybeCompact() {
	// Your Code Here (2C).
}

// allEntries return all the entries not compacted.
// note, exclude any dummy entries from the return value.
// note, this is one of the test stub functions you need to implement.
func (l *RaftLog) allEntries() []pb.Entry {
	// Your Code Here (2A).
	return l.entries[1:]
}

// unstableEntries return all the unstable entries
func (l *RaftLog) unstableEntries() []pb.Entry {
	// Your Code Here (2A).
	if l.stabled+1 >= uint64(len(l.entries)) {
		return make([]pb.Entry, 0)
	}
	return l.entries[l.stabled+1:]
}

// nextEnts returns all the committed but not applied entries
func (l *RaftLog) nextEnts() (ents []pb.Entry) {
	l.rwLock.RLock()
	defer l.rwLock.RUnlock()

	// Your Code Here (2A).
	if l.applied > l.committed {
		panic(fmt.Sprintf("[RaftLog] applied cannot be greater than committed. applied = %d, committed = %d", l.applied, l.committed))
	}
	if l.applied+1 >= uint64(len(l.entries)) {
		return make([]pb.Entry, 0)
	}
	// 以下逻辑注意切片截取是左闭右开区间
	// l.applied + 1 < len
	if l.committed+1 > uint64(len(l.entries)) {

		response := l.entries[l.applied+1:]
		l.DebugApplied("all committed", response)
		return response
	}
	// l.committed + 1 <= len
	response := l.entries[l.applied+1 : l.committed+1]
	l.DebugApplied("applied -> committed", response)
	return response
}

// LastIndex return the last index of the log entries
func (l *RaftLog) LastIndex() uint64 {
	// Your Code Here (2A).

	// 常规场景，最后一个Index=第一个元素的Index + 数组长度 - 1
	return l.entries[0].Index + uint64(len(l.entries)) - 1
}

// Term return the term of the entry in the given index
func (l *RaftLog) Term(i uint64) (uint64, error) {
	// Your Code Here (2A).
	return l.entries[i].Term, nil
}

func (l *RaftLog) nextSendEntries(nextIndex uint64) (entries []pb.Entry) {
	if nextIndex >= uint64(len(l.entries)) {
		return make([]pb.Entry, 0)
	}
	return l.entries[nextIndex:]
}

func (l *RaftLog) append(entry pb.Entry) {
	l.rwLock.Lock()
	defer l.rwLock.Unlock()

	if entry.Index <= l.LastIndex() {
		if l.entries[entry.Index].Term == entry.Term {
			return
		}
		// 截断
		l.entries = l.entries[0:entry.Index]
		if entry.Index <= l.stabled {
			l.stabled = entry.Index - 1
		}
	}

	if entry.Index < l.LastIndex()+1 {
		panic("[RaftLog] invalidate situation")
	} else if entry.Index == l.LastIndex()+1 {
		// append
		l.entries = append(l.entries, entry)
	} else {
		panic("[RaftLog] invalidate index")
	}
}

func (l *RaftLog) commit(idx uint64) {
	l.rwLock.Lock()
	defer l.rwLock.Unlock()
	if idx <= l.committed {
		// 这里会有晚到的请求，此时不能降低commit的值
		return
	} else {
		l.committed = idx
		log.Infof("[RaftLog][id = %d] set committed = %d", l.id, idx)
	}
}

func (l *RaftLog) logAtIndex(index uint64) (prevEntry *pb.Entry) {
	if index < uint64(len(l.entries)) {
		return &l.entries[index]
	} else {
		// TODO 这里偶现panic报错，待排查
		panic(fmt.Sprintf("[RaftLog] logAtIndex: invalid index: %d", index))
	}
}

func (l *RaftLog) DebugApplied(comment string, entries []pb.Entry) {
	if len(entries) == 0 {
		log.Infof("[RaftLog] %s. applied = %d, committed = %d, Empty. response = %+v", comment, l.applied, l.committed, entries)
	} else {
		log.Infof("[RaftLog] %s. applied = %d, committed = %d, last_index = %+v, response = %+v",
			comment, l.applied, l.committed, entries[len(entries)-1].Index, entries)
	}
}
